package com.cmnit.gatherdata.modules.module

import java.util.Calendar
import com.alibaba.fastjson.JSON
import com.cmnit.gatherdata.enums.KafkaTopicEnum
import com.cmnit.gatherdata.modules.bean.GantryBillHourSum
import com.cmnit.gatherdata.modules.utils.{DateUtil, KafkaStreamUtil}
import com.cmnit.gatherdata.utils.ConfigurationManager
import org.apache.log4j.Logger
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{Seconds, StreamingContext}

object GantryBillHourSumModule {

  final val logger = Logger.getLogger(GantryBillHourSumModule.getClass)

  def startStreaming(sparkSession: SparkSession, kafkaBootstrapServers: String, zkConnect: String, groupId: String): Unit = {

    val topic = KafkaTopicEnum.GBUPLOAD_ETCSU_TOPIC.getTopic
    val sc: SparkContext = sparkSession.sparkContext
    val ssc: StreamingContext = new StreamingContext(sc, Seconds(ConfigurationManager.getProperty("dual.time").toLong))

    // 获取kafka的实时流对象
    val dataInputStream = KafkaStreamUtil.getStream(ssc, kafkaBootstrapServers, zkConnect, topic, groupId)

    import sparkSession.implicits._
    dataInputStream.foreachRDD(rdd => {
      try {
        val df = rdd.repartition(2)
          .filter(x => !x.value.isEmpty)
          .map(_.value)
          .map(line => {
            JSON.parseObject(line.toString, classOf[GantryBillHourSum])
          }).map(info => {
          val calendar = Calendar.getInstance()
          // TODO 修改时间字段
          val time = DateUtil.parse(info.receivetime.toString, "yyyy-MM-dd HH:mm:ss")
          calendar.setTime(time)
          info.year = calendar.get(Calendar.YEAR).toString
          info.month = if ((calendar.get(Calendar.MONTH) + 1).toString.length < 2) {
            "0" + (calendar.get(Calendar.MONTH) + 1)
          } else {
            (calendar.get(Calendar.MONTH) + 1).toString
          }
          info.day = if (calendar.get(Calendar.DATE).toString.length < 2) {
            "0" + calendar.get(Calendar.DATE)
          } else {
            calendar.get(Calendar.DATE).toString
          }
          info
        }).toDF()
        df.show(30, truncate = false)
        // 创建视图
        df.createOrReplaceTempView("temp_GantryBillHourSum")
        sparkSession.sql("insert into table ods.ods_etc_gantrybillhoursum partition(year,month,day) select * from temp_GantryBillHourSum")

        // 维护本地offset
        KafkaStreamUtil.updateOffset(rdd, groupId)
      } catch {
        case e: Exception => println(e)
      }
    })

    ssc.start()
    ssc.awaitTermination()
  }
}
